package com.surfilter.massdata.spark.task.statistics;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;

import com.surfilter.massdata.spark.bean.StatDomain;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.JdbcUtil;

public class DomainCountStatisticsTask {

	public static void main(String[] args) {
		try{
			long start=System.currentTimeMillis();
			long temp=start;
			long temp_end=start;
			double exectime=0.0;
			
			String hiveDB = ConfigUtil.get("hiveDB", "myhive");
			SparkConf conf = new SparkConf().setAppName("DomainCountStatisticsTask");
			JavaSparkContext context = new JavaSparkContext(conf);
			HiveContext hiveContext = new HiveContext(context);
			
			String dayStr = ""; 
			if(args.length > 0){
				//传入日期格式：20170703
				dayStr = args[0];
			}
			//yyyyMMdd
			Date date = DateUtil.getExecDate1(DateUtil.getYesterday(), dayStr);
			String day = DateUtil.getCurrentymd(date);//2017-04-12
			System.out.println("统计数据时间："+DateUtil.getCurrentymd(date));
			
			String day_sql="select domain,top_domain  from "
					+ hiveDB + "." + "day_domain"
					+ " where ds='" + day +"' group by domain,top_domain ";
			System.out.println("day_sql:"+day_sql);
			
			JavaRDD<StatDomain> day_rdd=hiveContext.sql(day_sql).toJavaRDD().map(new StatDomainMap());
			DataFrame day_domainDF = hiveContext.createDataFrame(day_rdd,	StatDomain.class);
			day_domainDF.registerTempTable("t_day_domain");
			day_domainDF.persist(StorageLevel.MEMORY_AND_DISK_SER());
			
			insertWeekDomain(hiveContext,hiveDB,date);
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("insertWeekDomain exectime: "+exectime+" min........................");
			
			insertMonthDomain(hiveContext,hiveDB,date);
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("insertMonthDomain exectime: "+exectime+" min........................");
			
			insertQuarterDomain(hiveContext,hiveDB,date);
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("insertQuarterDomain exectime: "+exectime+" min........................");
			
			insertYearDomain(hiveContext,hiveDB,date);
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("insertYearDomain exectime: "+exectime+" min........................");
			
			long end=System.currentTimeMillis();
			exectime=(end-start)*1.0/(1000*60);
			System.out.println("DomainCountStatisticsTask exectime: "+exectime+" min........................");
			
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	private static void insertYearDomain(HiveContext hiveContext,
			String hiveDB, Date date) {
		Connection conn = null;
		try{
			String year = DateUtil.getCurrentYear_s(date);
			String day = DateUtil.getCurrentymd(date);//2017-04-12
			//取前一天时间
			Date yesterday = new Date(date.getTime()-24*60*60*1000);
			String year_last = DateUtil.getCurrentYear_s(yesterday);
			String day_last = DateUtil.getCurrentymd(yesterday);
			//网站数目
			long website_count=0l;
			//域名数目
			long domain_count=0l;
			//判断是否为新的季节
			if(!year.equals(year_last)){
				String init_sql="insert into table "+ hiveDB + ".dws_domain_year partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+" from t_day_domain ";
				System.out.println("year_init_sql:"+init_sql);
				hiveContext.sql(init_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_day_domain";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_day_domain";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}else{
				String sql="select domain,top_domain  from "
						+ hiveDB + "." + "dws_domain_year"
						+ " where ds='" + day_last +"'";
				System.out.println("year_sql:"+sql);
				JavaRDD<Row> year_rdd=hiveContext.sql(sql).toJavaRDD();
				
				JavaRDD<StatDomain> year_domain_rdd =year_rdd.map(new StatDomainMap());
				DataFrame DomainDF = hiveContext.createDataFrame(year_domain_rdd,	StatDomain.class);
				DomainDF.registerTempTable("t_domain_year");
				//昨天的dws_domain_year和day_domain 数据 union
				String sql_union="select domain,top_domain from ( select domain,top_domain from t_domain_year union select domain,top_domain from t_day_domain) t group by domain,top_domain ";
				JavaRDD<StatDomain> union_rdd=hiveContext.sql(sql_union).toJavaRDD().map(new StatDomainMap());
				DataFrame UnionDF = hiveContext.createDataFrame(union_rdd,	StatDomain.class);
				UnionDF.registerTempTable("t_domain_year_union");
				
				String year_insert_sql="insert into table "+ hiveDB + ".dws_domain_year partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+" from t_domain_year_union ";
				System.out.println("year_insert_sql:"+year_insert_sql);
				hiveContext.sql(year_insert_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_domain_year_union";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_domain_year_union";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}
			conn = JdbcUtil.getConnection();
			//插入oracle
			//website
			String columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			String values="values(SA5100_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> websiteMap= new HashMap<Integer,Object>(); 		
			websiteMap.put(1, DateUtil.getCurrentYear(date));//year
			websiteMap.put(2, DateUtil.getHalfYear(date));//half_year
			websiteMap.put(3, DateUtil.getQuarter(date));//quarter
			websiteMap.put(4, DateUtil.getCurrentMonth(date));//month
			websiteMap.put(5, DateUtil.getCurrentWeek(date));//week
			websiteMap.put(6, DateUtil.getCurrentDay(date));//day
			websiteMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			websiteMap.put(8, "1");//buss_value
			websiteMap.put(9, website_count);//count_value
			websiteMap.put(10, 0);//sta_range
		
			JdbcUtil.deleteYearTable(date,conn,"SA5100","WEBSITE_COUNT_SUM","1",0);
			JdbcUtil.insertTable_oneline(conn,websiteMap,"SA5100",columns,values);
			//domain
			columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			values="values(SA6100_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> domainMap= new HashMap<Integer,Object>(); 		
			domainMap.put(1, DateUtil.getCurrentYear(date));//year
			domainMap.put(2, DateUtil.getHalfYear(date));//half_year
			domainMap.put(3, DateUtil.getQuarter(date));//quarter
			domainMap.put(4, DateUtil.getCurrentMonth(date));//month
			domainMap.put(5, DateUtil.getCurrentWeek(date));//week
			domainMap.put(6, DateUtil.getCurrentDay(date));//day
			domainMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			domainMap.put(8, "");//buss_value
			domainMap.put(9, domain_count);//count_value
			domainMap.put(10, 0);//sta_range
			JdbcUtil.deleteYearTable(date,conn,"SA6100","WEBSITE_COUNT_SUM","",0);
			JdbcUtil.insertTable_oneline(conn,domainMap,"SA6100",columns,values);
			
		}catch(Exception e){
			if(conn!=null){
				JdbcUtil.close(conn);
			}
			e.printStackTrace();
		}finally{
			if(conn!=null){
				JdbcUtil.close(conn);
			}
		}
	
		
	
		
	}

	private static void insertQuarterDomain(HiveContext hiveContext,
			String hiveDB, Date date) {


		Connection conn = null;
		try{
			String year = DateUtil.getCurrentYear_s(date);
			int quarter = DateUtil.getQuarter(date);
			String day = DateUtil.getCurrentymd(date);//2017-04-12
			//取前一天时间
			Date yesterday = new Date(date.getTime()-24*60*60*1000);
			int quarter_last = DateUtil.getQuarter(yesterday);
			String day_last = DateUtil.getCurrentymd(yesterday);
			//网站数目
			long website_count=0l;
			//域名数目
			long domain_count=0l;
			//判断是否为新的季节
			if(quarter!=quarter_last){
				String init_sql="insert into table "+ hiveDB + ".dws_domain_quarter partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+quarter+" from t_day_domain ";
				System.out.println("quarter_init_sql:"+init_sql);
				hiveContext.sql(init_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_day_domain";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_day_domain";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}else{
				String sql="select domain,top_domain  from "
						+ hiveDB + "." + "dws_domain_quarter"
						+ " where ds='" + day_last +"'";
				System.out.println("quarter_sql:"+sql);
				JavaRDD<Row> quarter_rdd=hiveContext.sql(sql).toJavaRDD();
				
				JavaRDD<StatDomain> quarter_domain_rdd =quarter_rdd.map(new StatDomainMap());
				DataFrame DomainDF = hiveContext.createDataFrame(quarter_domain_rdd,	StatDomain.class);
				DomainDF.registerTempTable("t_domain_quarter");
				//昨天的dws_domain_quarter和day_domain 数据 union
				String sql_union="select domain,top_domain from ( select domain,top_domain from t_domain_quarter union select domain,top_domain from t_day_domain) t group by domain,top_domain ";
				JavaRDD<StatDomain> union_rdd=hiveContext.sql(sql_union).toJavaRDD().map(new StatDomainMap());
				DataFrame UnionDF = hiveContext.createDataFrame(union_rdd,	StatDomain.class);
				UnionDF.registerTempTable("t_domain_quarter_union");
				
				String quarter_insert_sql="insert into table "+ hiveDB + ".dws_domain_quarter partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+quarter+" from t_domain_quarter_union ";
				System.out.println("quarter_insert_sql:"+quarter_insert_sql);
				hiveContext.sql(quarter_insert_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_domain_quarter_union";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_domain_quarter_union";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}
			conn = JdbcUtil.getConnection();
			//插入oracle
			//website
			String columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			String values="values(SA5301_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> websiteMap= new HashMap<Integer,Object>(); 		
			websiteMap.put(1, DateUtil.getCurrentYear(date));//year
			websiteMap.put(2, DateUtil.getHalfYear(date));//half_year
			websiteMap.put(3, DateUtil.getQuarter(date));//quarter
			websiteMap.put(4, DateUtil.getCurrentMonth(date));//month
			websiteMap.put(5, DateUtil.getCurrentWeek(date));//week
			websiteMap.put(6, DateUtil.getCurrentDay(date));//day
			websiteMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			websiteMap.put(8, "1");//buss_value
			websiteMap.put(9, website_count);//count_value
			websiteMap.put(10, 0);//sta_range
		
			JdbcUtil.deleteQuarterTable(date,conn,"SA5301","WEBSITE_COUNT_SUM","1",0);
			JdbcUtil.insertTable_oneline(conn,websiteMap,"SA5301",columns,values);
			//domain
			columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			values="values(SA6300_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> domainMap= new HashMap<Integer,Object>(); 		
			domainMap.put(1, DateUtil.getCurrentYear(date));//year
			domainMap.put(2, DateUtil.getHalfYear(date));//half_year
			domainMap.put(3, DateUtil.getQuarter(date));//quarter
			domainMap.put(4, DateUtil.getCurrentMonth(date));//month
			domainMap.put(5, DateUtil.getCurrentWeek(date));//week
			domainMap.put(6, DateUtil.getCurrentDay(date));//day
			domainMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			domainMap.put(8, "");//buss_value
			domainMap.put(9, domain_count);//count_value
			domainMap.put(10, 0);//sta_range
			JdbcUtil.deleteQuarterTable(date,conn,"SA6300","WEBSITE_COUNT_SUM","",0);
			JdbcUtil.insertTable_oneline(conn,domainMap,"SA6300",columns,values);
			
		}catch(Exception e){
			if(conn!=null){
				JdbcUtil.close(conn);
			}
			e.printStackTrace();
		}finally{
			if(conn!=null){
				JdbcUtil.close(conn);
			}
		}
	
		
	}

	private static void insertMonthDomain(HiveContext hiveContext,
			String hiveDB, Date date) {

		Connection conn = null;
		try{
			String year = DateUtil.getCurrentYear_s(date);
			String month = DateUtil.getCurrentMonth_s(date);
			String day = DateUtil.getCurrentymd(date);//2017-04-12
			//取前一天时间
			Date yesterday = new Date(date.getTime()-24*60*60*1000);
			String month_last = DateUtil.getCurrentMonth_s(yesterday);
			String day_last = DateUtil.getCurrentymd(yesterday);
			//网站数目
			long website_count=0l;
			//域名数目
			long domain_count=0l;
			//判断是否为新的月份
			if(!month_last.equals(month)){
				String init_sql="insert into table "+ hiveDB + ".dws_domain_month partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+month+" from t_day_domain ";
				System.out.println("month_init_sql:"+init_sql);
				hiveContext.sql(init_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_day_domain";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_day_domain";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}else{
				String sql="select domain,top_domain  from "
						+ hiveDB + "." + "dws_domain_month"
						+ " where ds='" + day_last +"'";
				System.out.println("month_sql:"+sql);
				JavaRDD<Row> month_rdd=hiveContext.sql(sql).toJavaRDD();
				
				JavaRDD<StatDomain> month_domain_rdd =month_rdd.map(new StatDomainMap());
				DataFrame DomainDF = hiveContext.createDataFrame(month_domain_rdd,	StatDomain.class);
				DomainDF.registerTempTable("t_domain_month");
				//昨天的dws_domain_month和day_domain 数据 union
				String sql_union="select domain,top_domain from ( select domain,top_domain from t_domain_month union select domain,top_domain from t_day_domain) t group by domain,top_domain ";
				JavaRDD<StatDomain> union_rdd=hiveContext.sql(sql_union).toJavaRDD().map(new StatDomainMap());
				DataFrame UnionDF = hiveContext.createDataFrame(union_rdd,	StatDomain.class);
				UnionDF.registerTempTable("t_domain_month_union");
				
				String month_insert_sql="insert into table "+ hiveDB + ".dws_domain_month partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+month+" from t_domain_month_union ";
				System.out.println("month_insert_sql:"+month_insert_sql);
				hiveContext.sql(month_insert_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_domain_month_union";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_domain_month_union";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}
			conn = JdbcUtil.getConnection();
			//插入oracle
			//website
			String columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			String values="values(SA5401_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> websiteMap= new HashMap<Integer,Object>(); 		
			websiteMap.put(1, DateUtil.getCurrentYear(date));//year
			websiteMap.put(2, DateUtil.getHalfYear(date));//half_year
			websiteMap.put(3, DateUtil.getQuarter(date));//quarter
			websiteMap.put(4, DateUtil.getCurrentMonth(date));//month
			websiteMap.put(5, DateUtil.getCurrentWeek(date));//week
			websiteMap.put(6, DateUtil.getCurrentDay(date));//day
			websiteMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			websiteMap.put(8, "1");//buss_value
			websiteMap.put(9, website_count);//count_value
			websiteMap.put(10, 0);//sta_range
		
			JdbcUtil.deleteMonthTable(date,conn,"SA5401","WEBSITE_COUNT_SUM","1",0);
			JdbcUtil.insertTable_oneline(conn,websiteMap,"SA5401",columns,values);
			//domain
			columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			values="values(SA6401_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> domainMap= new HashMap<Integer,Object>(); 		
			domainMap.put(1, DateUtil.getCurrentYear(date));//year
			domainMap.put(2, DateUtil.getHalfYear(date));//half_year
			domainMap.put(3, DateUtil.getQuarter(date));//quarter
			domainMap.put(4, DateUtil.getCurrentMonth(date));//month
			domainMap.put(5, DateUtil.getCurrentWeek(date));//week
			domainMap.put(6, DateUtil.getCurrentDay(date));//day
			domainMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			domainMap.put(8, "");//buss_value
			domainMap.put(9, domain_count);//count_value
			domainMap.put(10, 0);//sta_range
			JdbcUtil.deleteMonthTable(date,conn,"SA6401","WEBSITE_COUNT_SUM","",0);
			JdbcUtil.insertTable_oneline(conn,domainMap,"SA6401",columns,values);
			
		}catch(Exception e){
			if(conn!=null){
				JdbcUtil.close(conn);
			}
			e.printStackTrace();
		}finally{
			if(conn!=null){
				JdbcUtil.close(conn);
			}
		}
	}

	private static void insertWeekDomain(HiveContext hiveContext,String hiveDB,Date date){
		Connection conn = null;
		try{
			String year = DateUtil.getCurrentYear_s(date);
			String week = DateUtil.getCurrentWeek_s(date);
			String day = DateUtil.getCurrentymd(date);//2017-04-12
			//取前一天时间
			Date yesterday = new Date(date.getTime()-24*60*60*1000);
			String week_last = DateUtil.getCurrentWeek_s(yesterday);
			String day_last = DateUtil.getCurrentymd(yesterday);
			//网站数目
			long website_count=0l;
			//域名数目
			long domain_count=0l;
			//判断是否为新一周
			if(!week_last.equals(week)){
				String week_init_sql="insert into table "+ hiveDB + ".dws_domain_week partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+week+" from t_day_domain ";
				System.out.println("week_init_sql:"+week_init_sql);
				hiveContext.sql(week_init_sql);
				//网站数量
				String website_num_sql="select count( distinct top_domain) as count_value  from t_day_domain";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_day_domain";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}else{
				String week_sql="select domain,top_domain  from "
						+ hiveDB + "." + "dws_domain_week"
						+ " where ds='" + day_last +"'";
				System.out.println("week_sql:"+week_sql);
				JavaRDD<Row> week_rdd=hiveContext.sql(week_sql).toJavaRDD();
				
				JavaRDD<StatDomain> week_domain_rdd =week_rdd.map(new StatDomainMap());
				DataFrame DomainDF = hiveContext.createDataFrame(week_domain_rdd,	StatDomain.class);
				DomainDF.registerTempTable("t_domain_week");
				//昨天的dws_domain_week和day_domain 数据 union
				String sql_union="select domain,top_domain from ( select domain,top_domain from t_domain_week union select domain,top_domain from t_day_domain) t group by domain,top_domain ";
				JavaRDD<StatDomain> union_rdd=hiveContext.sql(sql_union).toJavaRDD().map(new StatDomainMap());
				DataFrame UnionDF = hiveContext.createDataFrame(union_rdd,	StatDomain.class);
				UnionDF.registerTempTable("t_domain_week_union");
				
				String week_insert_sql="insert into table "+ hiveDB + ".dws_domain_week partition (ds='"+day+"')"
						+ " select domain,top_domain,"+year+","+week+" from t_domain_week_union ";
				System.out.println("week_insert_sql:"+week_insert_sql);
				hiveContext.sql(week_insert_sql);
				//网站数量
				String website_num_sql="select count(distinct top_domain) as count_value  from t_domain_week_union";
				List<Row> list_website=hiveContext.sql(website_num_sql).toJavaRDD().collect();
				
				if(list_website.size()>0){
					website_count=Long.valueOf(list_website.get(0).getAs("count_value").toString());
				}
				//域名数量
				String domain_num_sql="select count(domain) as count_value from t_domain_week_union";
				List<Row> list_domain=hiveContext.sql(domain_num_sql).toJavaRDD().collect();
				if(list_domain.size()>0){
					domain_count=Long.valueOf(list_domain.get(0).getAs("count_value").toString());
				}
				
			}
			conn = JdbcUtil.getConnection();
			//插入oracle
			//website
			String columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			String values="values(SA5501_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> websiteMap= new HashMap<Integer,Object>(); 		
			websiteMap.put(1, DateUtil.getCurrentYear(date));//year
			websiteMap.put(2, DateUtil.getHalfYear(date));//half_year
			websiteMap.put(3, DateUtil.getQuarter(date));//quarter
			websiteMap.put(4, DateUtil.getCurrentMonth(date));//month
			websiteMap.put(5, DateUtil.getCurrentWeek(date));//week
			websiteMap.put(6, DateUtil.getCurrentDay(date));//day
			websiteMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			websiteMap.put(8, "1");//buss_value
			websiteMap.put(9, website_count);//count_value
			websiteMap.put(10, 0);//sta_range
		
			JdbcUtil.deleteWeekTable(date,conn,"SA5501","WEBSITE_COUNT_SUM","1",0);
			JdbcUtil.insertTable_oneline(conn,websiteMap,"SA5501",columns,values);
			//domain
			columns="id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range";
			values="values(SA6501_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
			Map<Integer,Object> domainMap= new HashMap<Integer,Object>(); 		
			domainMap.put(1, DateUtil.getCurrentYear(date));//year
			domainMap.put(2, DateUtil.getHalfYear(date));//half_year
			domainMap.put(3, DateUtil.getQuarter(date));//quarter
			domainMap.put(4, DateUtil.getCurrentMonth(date));//month
			domainMap.put(5, DateUtil.getCurrentWeek(date));//week
			domainMap.put(6, DateUtil.getCurrentDay(date));//day
			domainMap.put(7, "WEBSITE_COUNT_SUM");//buss_type
			domainMap.put(8, "");//buss_value
			domainMap.put(9, domain_count);//count_value
			domainMap.put(10, 0);//sta_range
			JdbcUtil.deleteWeekTable(date,conn,"SA6501","WEBSITE_COUNT_SUM","",0);
			JdbcUtil.insertTable_oneline(conn,domainMap,"SA6501",columns,values);
			
		}catch(Exception e){
			if(conn!=null){
				JdbcUtil.close(conn);
			}
			e.printStackTrace();
		}finally{
			if(conn!=null){
				JdbcUtil.close(conn);
			}
			
		}
	}
}
class StatDomainMap implements Function<Row, StatDomain>{

	@Override
	public StatDomain call(Row v1) throws Exception {
		// TODO Auto-generated method stub
		StatDomain e =new StatDomain();
		e.setDomain(v1.getAs("domain").toString());
		e.setTop_domain(v1.getAs("top_domain").toString());
		return e;
	}

}

